package report

import java.util.Properties

import Configer.Config
import org.apache.commons.lang.StringUtils
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.SQLContext
import org.apache.spark.{SparkConf, SparkContext}
import util.{JedisPools, KpiPublic}

//用redis的方式分析app
object AppKpiRedis {
  def main(args: Array[String]): Unit = {
    //sparkContext
    val conf = new SparkConf()
    conf.setMaster("local[*]")
    conf.setAppName(s"${this.getClass.getName}")
    conf.set("spark.serializer",Config.serializer)

    val sc = new SparkContext(conf)
    val sQLContext = new SQLContext(sc)
    val dataFrame = sQLContext.read.parquet(Config.parquetPath)
    val resultRDD: RDD[(String, List[Double])] = dataFrame.mapPartitions(partition => {
      val jedis = JedisPools.getJedis(15)
      val tuples = partition.map(row => {
        var appName = row.getAs[String]("appname")
        val appId = row.getAs[String]("appid")

        if (StringUtils.isEmpty(appName)) {
          if (StringUtils.isNotEmpty(appId)) {
            appName = jedis.hget("appdict", appId)
            if (StringUtils.isEmpty(appName)) {
              appName = appId
            }
          } else {
            appName = "某app"
          }
        }
        (appName, KpiPublic.KpiPublic(row))
      })
      jedis.close()
      tuples
    }).reduceByKey((l1, l2) => l1 zip l2 map (t => t._1 + t._2))
    resultRDD

    val props = new Properties()
    props.setProperty("driver",Config.driver)
    props.setProperty("user",Config.user)
    props.setProperty("password",Config.password)
    import sQLContext.implicits._
    resultRDD.map(row=>(row._1,row._2(0),row._2(1),row._2(2),row._2(3),row._2(4),row._2(7),row._2(8),row._2(5),row._2(6)))
      .toDF("appname","adOldReq","adEffReq","adReq","adRtbReq","adSucReq","adshow","adclick","adadpReq","adCusReq")
      .write.jdbc(Config.url,Config.table,props)

    sc.stop()
  }
}
